package com.yanggu.flink.datastream_api.test

import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses
import org.apache.flink.util.OutputTag
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers

/**
 * 简单测试ProcessFunction, 没有状态和定时器
 * 有侧输出流
 */
class TestPassThroughProcessFunction extends AnyFlatSpec with Matchers {

  "PassThroughProcessFunction" should "forward values" in {
    val processFunction = new PassThroughProcessFunction()

    val harness = ProcessFunctionTestHarnesses.forProcessFunction(processFunction)

    //发送数据
    harness.processElement(1, 10)

    //验证是否向下游输出
    harness.extractOutputValues().get(0) shouldBe 1

    //验证侧输出流是否输出
    harness.getSideOutput(new OutputTag[Integer]("test-output-tag", Types.INT)).poll().getValue shouldBe 1

  }

}
